package com.atguigu.flinksql.day12;

import com.atguigu.datastream.bean.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * ClassName: Test01
 * Package: com.atguigu.flinksql.day12
 * Description:
 *  1.1 使用FlinkSQL读取Kafka数据(json:id,vc,ts)
 * 	1.2 分别使用GroupWindow和TVF方式实现  处理时间语义下  10s滚动窗口计算每个传感器(id)的最高水位(vc)
 * 	1.3 转换为流进行打印
 * @Author ChenJun
 * @Create 2023/4/21 8:34
 * @Version 1.0
 */
public class Test01 {
    public static void main(String[] args) throws Exception {

        //1. 创建流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2. 从kafka读取数据
        tableEnv.executeSql(""+
                "CREATE TABLE t1_pt( \n" +
                "    id string, \n" +
                "    ts bigint, \n" +
                "    vc int,\n" +
                "    pt AS PROCTIME()\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                "  'properties.group.id' = 'test1' ,\n" +
                "  'scan.startup.mode' = 'group-offsets' ,\n" +
                "  'sink.partitioner' = 'fixed',\n" +
                "  'topic' = 'test1109',\n" +
                "  'format' = 'json'\n" +
                ")");

        //3.1 分别使用GroupWindow和处理时间语义下  10s滚动窗口计算每个传感器(id)的最高水位(vc)
        Table table = tableEnv.sqlQuery("" +
                "select\n" +
                "    TUMBLE_START(pt, INTERVAL '10' SECOND) AS wStart,\n" +
                "    TUMBLE_END(pt, INTERVAL '10' SECOND) AS wEnd,\n" +
                "    id,\n" +
                "    max(vc) max_vc\n" +
                "from t1_pt\n" +
                "group by id,TUMBLE(pt, INTERVAL '10' SECOND)");

        //转换为流
        DataStream<Tuple2<Boolean, Row>> dataStream = tableEnv.toRetractStream(table, Row.class);
        //打印
        dataStream.print("GroupWindow->");

        //3.2 分别使用TVF方式实现  处理时间语义下  10s滚动窗口计算每个传感器(id)的最高水位(vc)
        Table table1 = tableEnv.sqlQuery("" +
                "SELECT \n" +
                "   window_start,\n" +
                "   window_end, \n" +
                "   max(vc) as max_vc\n" +
                "FROM TABLE(\n" +
                "    TUMBLE(TABLE t1_pt, DESCRIPTOR(pt), INTERVAL '10' second))\n" +
                "GROUP BY \n" +
                "\t  window_start, \n" +
                "\t  window_end");

        //转为流
        DataStream<Tuple2<Boolean, Row>> dataStream1 = tableEnv.toRetractStream(table1, Row.class);
        //打印
        dataStream1.print("VTF->");


        //4. 执行
        env.execute();

    }
}
